package com.parsechina.hpush.broker.internal;

import com.parsechina.hpush.broker.config.HPushConfig;
import com.parsechina.hpush.broker.spi.ProtocolProcessor;
import com.parsechina.hpush.broker.spi.ServerCycle;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.handler.codec.http.HttpContentCompressor;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author linfeng
 * @since 2016/11/22
 */
@Component
public class NettyServerCycle implements ServerCycle {

    private static final Logger LOG = LoggerFactory.getLogger(NettyServerCycle.class);
    private static final int DEFAULT_CONNECT_TIMEOUT = 10;
    private static final String MQTT_SUBPROTOCOL_CSV_LIST = "mqtt, mqttv3.1, mqttv3.1.1";
    private static final int DEFAULT_LENGTH = 268435455;

    private final EventLoopGroup bossGroup;
    private final EventLoopGroup workerGroup;

    private final ProtocolProcessor protocolProcessor;
    private final HPushConfig props;

    @Autowired
    public NettyServerCycle(EventLoopGroup bossGroup, EventLoopGroup workerGroup,
                            ProtocolProcessor protocolProcessor, HPushConfig config) {
        this.bossGroup = bossGroup;
        this.workerGroup = workerGroup;
        this.protocolProcessor = protocolProcessor;
        this.props = config;
    }


    @Override
    public void startup() {

        final NettyMqttHandler handler = new NettyMqttHandler(protocolProcessor);

        initializePlainTCPTransport(handler, props);
        initializeWebSocketTransport(handler, props);

    }

    @Override
    public void shutdown() {

        if (workerGroup == null) {
            throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
        }
        if (bossGroup == null) {
            throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
        }

        Future<?> workerWaiter = workerGroup.shutdownGracefully();
        Future<?> bossWaiter = bossGroup.shutdownGracefully();

        try {
            workerWaiter.await(100);
        } catch (InterruptedException iex) {
            throw new IllegalStateException(iex);
        }

        try {
            bossWaiter.await(100);
        } catch (InterruptedException iex) {
            throw new IllegalStateException(iex);
        }

    }


    interface ChannelPipelineInitializer {
        /**
         * 初始化netty处理管道
         *
         * @param pipeline ChannelPipeline
         * @throws Exception 异常
         */
        void init(ChannelPipeline pipeline) throws Exception;

    }

    static class WebSocketFrameToByteBufDecoder extends MessageToMessageDecoder<BinaryWebSocketFrame> {

        @Override
        protected void decode(ChannelHandlerContext chc, BinaryWebSocketFrame frame, List<Object> out) throws Exception {
            //convert the frame to a ByteBuf
            ByteBuf bb = frame.content();
            bb.retain();
            out.add(bb);
        }
    }

    static class ByteBufToWebSocketFrameEncoder extends MessageToMessageEncoder<ByteBuf> {

        @Override
        protected void encode(ChannelHandlerContext chc, ByteBuf bb, List<Object> out) throws Exception {
            //convert the ByteBuf to a WebSocketFrame
            BinaryWebSocketFrame result = new BinaryWebSocketFrame();
            result.content().writeBytes(bb);
            out.add(result);
        }
    }

    //==================================================================================================================


    /**
     * 通用初始化处理
     *
     * @param host        服务器host
     * @param port        端口
     * @param initializer netty处理管道初始化接口
     */
    private void initHandlerFactory(String host, int port, final ChannelPipelineInitializer initializer) {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        try {
                            initializer.init(pipeline);
                        } catch (Throwable th) {
                            LOG.error("Severe error during pipeline creation", th);
                            throw th;
                        }
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 128)
                .option(ChannelOption.SO_REUSEADDR, true)
                //连接超时30000毫秒
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,30000)
                //输入流的read方法被阻塞时，接受数据的等待超时时间5000毫秒，
                .childOption(ChannelOption.SO_TIMEOUT,5000)
                //缓冲池
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_KEEPALIVE, true);
        try {
            ChannelFuture bind = b.bind(host, port);
            LOG.info("HPush server bind host: {}, port: {}", host, port);
            bind.sync();
        } catch (InterruptedException ex) {
            LOG.error(null, ex);
        }
    }

    /**
     * 初始化TCP传输处理
     *
     * @param handler mqtt处理
     * @param props   配置
     */
    private void initializePlainTCPTransport(final NettyMqttHandler handler, final HPushConfig props) {
        final NettyIdleTimeoutHandler timeoutHandler = new NettyIdleTimeoutHandler();
        final String host = props.getHost();
        final int port = props.getPort();
        initHandlerFactory(host, port, pipeline -> {
            pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 0, DEFAULT_CONNECT_TIMEOUT));
            pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
            pipeline.addLast("decoder", new MqttDecoder(DEFAULT_LENGTH));
            pipeline.addLast("encoder", MqttEncoder.INSTANCE);
            pipeline.addLast("handler", handler);
        });
    }

    /**
     * 初始化websocket传输
     *
     * @param handler
     * @param props
     */
    private void initializeWebSocketTransport(final NettyMqttHandler handler, final HPushConfig props) {

        final NettyIdleTimeoutHandler timeoutHandler = new NettyIdleTimeoutHandler();
        final int port = props.getWebsocketPort();
        final String host = props.getHost();
        initHandlerFactory(host, port, pipeline -> {
            pipeline.addLast("httpEncoder", new HttpResponseEncoder());
            pipeline.addLast("httpDecoder", new HttpRequestDecoder());
            pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
            pipeline.addLast("deflater", new HttpContentCompressor());
            pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
            pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
            pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
            pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, DEFAULT_CONNECT_TIMEOUT));
            pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
            pipeline.addLast("decoder", new MqttDecoder(DEFAULT_LENGTH));
            pipeline.addLast("encoder", MqttEncoder.INSTANCE);
            pipeline.addLast("handler", handler);
        });
    }

}
